package com.lc1993929.mr.join;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by LiuChang on 2017/12/14/014.
 */
public class JoinReducer extends Reducer<Text, JoinBean, Text, JoinBean> {

    @Override
    protected void reduce(Text key, Iterable<JoinBean> values, Context context) throws IOException, InterruptedException {
        List<JoinBean> orderJoinBeans = new ArrayList<>();
        List<JoinBean> userJoinBeans = new ArrayList<>();

        values.forEach(joinBean -> {
            String tableName = joinBean.getTableName();
            if ("order.txt".equals(tableName)) {
                orderJoinBeans.add(joinBean);
            } else {
                userJoinBeans.add(joinBean);
            }
        });

        for (JoinBean userJoinBean : userJoinBeans) {
            for (JoinBean orderJoinBean : orderJoinBeans) {
                orderJoinBean.setUname(userJoinBean.getUname());
                orderJoinBean.setAge(userJoinBean.getAge());
                orderJoinBean.setLover(userJoinBean.getLover());
                context.write(key, orderJoinBean);
            }
        }

    }
}
